package cn.getech.data.development.controller;


import cn.getech.data.development.config.properties.BdpJobConfig;
import cn.getech.data.development.config.properties.OozieConfig;
import cn.getech.data.development.constant.*;
import cn.getech.data.development.dto.IdAndNameDto;
import cn.getech.data.development.dto.JobNodeConfigDto;
import cn.getech.data.development.dto.JobNodeLayConfigDto;
import cn.getech.data.development.entity.*;
import cn.getech.data.development.model.res.jobInfo.JobDataRes;
import cn.getech.data.development.model.vo.CopyJobNodeInfoVO;
import cn.getech.data.development.model.vo.IdAndNameVO;
import cn.getech.data.development.model.vo.JobDataVo;
import cn.getech.data.development.model.vo.JobNodeConfigVo;
import cn.getech.data.development.service.*;
import cn.getech.data.development.utils.HdfsUtil;
import cn.getech.data.intelligence.common.exception.RRException;
import cn.getech.data.intelligence.common.utils.PojoUtils;
import cn.getech.data.intelligence.common.utils.R;
import cn.getech.system.center.annotation.SysLog;
import cn.getech.system.center.constant.OperationModulesEnum;
import cn.getech.system.center.constant.OperationTypeEnum;
import cn.getech.system.center.constant.module.ModuleEnum;
import cn.getech.system.center.entity.SysUserEntity;
import cn.getech.system.center.service.SysUserService;
import cn.getech.system.center.utils.ShiroUtils;
import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiParam;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.Assert;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;

import java.util.*;
import java.util.stream.Collectors;

/**
 * <p>
 * 任务节点信息表 前端控制器
 * </p>
 *
 * @author zenith
 * @since 2019-07-08
 */
@Slf4j
@Api(tags = "任务节点信息")
@RestController
@RequestMapping("/bdp/jobNodeInfo")
public class JobNodeInfoController {

    @Autowired
    private JobInfoService jobInfoService;
    @Autowired
    private JobNodeInfoService jobNodeInfoService;
    @Autowired
    private JobNodeConfService jobNodeConfService;
    @Autowired
    private JobNodeRunHistoryService jobNodeRunHistoryService;
    @Autowired
    private TableInfoService tableInfoService;
    @Autowired
    private TableFieldInfoService tableFieldInfoService;
    @Autowired
    private BdpJobConfig bdpJobConfig;
    @Autowired
    private OozieConfig oozieConfig;
    @Autowired
    private SysUserService sysUserService;
    @Autowired
    private BussessConfigService bussessConfigService;



    /**
     * 列表
     *
     * @param params
     * @return
     */
    @ApiOperation("列表")
    @GetMapping("/list")
   // @SysLog(value="列表",type= OperationTypeEnum.SELECTCODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
//    @RequiresPermissions("bdp:jobNodeInfo:list")
    public R list(@ApiParam(value = "{'job_id':1}") @RequestParam Map<String, Object> params) {
        String job_id = (String) params.get("job_id");
        return R.okWithData(jobNodeInfoService.list(
                new QueryWrapper<JobNodeInfo>().like(StringUtils.isNotBlank(job_id), "job_id", job_id)
                        .eq("create_per", ShiroUtils.getUserId().intValue())
        ));
    }


    @ApiOperation("保存数据任务节点配置")
    @PostMapping("/saveJobNodeConfig")
    //todo 0925
    //@RequiresPermissions({"bdp:procInfo:develop"})
    @SysLog(value="保存数据任务节点配置",type= OperationTypeEnum.UPDATECODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    public R saveJobNodeConfig(@ApiParam(value = "{'name':'test','nodeKey':'2','jobId':1,'typeId':1,'jobNodeConfig':{}}") @RequestBody JobNodeInfo param) throws Exception {
        Assert.notNull(param, DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
        jobNodeInfoService.saveJobNodeConfig(param);
        return R.ok();
    }


    @ApiOperation("复制数据任务节点配置")
    @PostMapping("/copySaveJobNodeConfig")
    //todo 0925
    //@RequiresPermissions({"bdp:procInfo:develop"})
    @SysLog(value="复制任务节点配置",type= OperationTypeEnum.UPDATECODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    public R copySaveJobNodeConfig(@ApiParam(value = "{'name':'test','nodeKey':'2','jobId':1,'typeId':1,'jobNodeConfig':{}}") @RequestBody CopyJobNodeInfoVO copyJobNodeInfoVO) throws Exception {
        Assert.notNull(copyJobNodeInfoVO, DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
        jobNodeInfoService.copySaveJobNodeConfig(copyJobNodeInfoVO);
        return R.ok();
    }



    @GetMapping("/nodelistOutTable/{jobNodeId}")
    //@SysLog(value="获取上游输出表",type= OperationTypeEnum.SELECTCODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    public R nodelistOutTable(@PathVariable("jobNodeId") Integer jobNodeId) {
        //根据jobNodeId查询节点对应的上游任务输出表信息
        List<IdAndNameDto> listRest = jobNodeInfoService.selectNodeLayConfig(jobNodeId);
        return R.okWithData(PojoUtils.listConvert(IdAndNameVO.class,listRest));
    }


    @GetMapping("/routeType")
    //@SysLog(value="获取路由信息",type= OperationTypeEnum.SELECTCODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    public R routeType() {
        //查询对应的路由的节点类型
        List<JSONObject> data = new ArrayList<>();
        for (RouteTypeEnum e : RouteTypeEnum.values()
        ) {
            JSONObject obj = new JSONObject();
            obj.put("id", e.getId());
            obj.put("name", e.getNameDes());
            obj.put("nameEn", e.getName());
            obj.put("ops",e.getOpters());
            data.add(obj);
        }
        return R.okWithData(data);
    }

//    /**
//     * 保存数据任务节点配置
//     */
//    @SysLog(value = "保存数据任务节点配置", type = OperationTypeEnum.ADDCODE, modulesType = OperationModulesEnum.JOBMODELCODE)
//    @ApiOperation("保存数据任务节点配置(除结束节点外)")
//    @PostMapping("/save")
//    //    @RequiresPermissions("bdp:jobNodeInfo:save")
//    public R save(@ApiParam(value = "{'name':'test','nodeKey':'2','jobId':1,'typeId':1,'jobNodeConfig':{}}") @RequestBody JobNodeInfo param) throws Exception {
//
//        //验证参数的数据
//        if (null == param || StringUtils.isEmpty(param.getNodeKey()) || null == param.getJobId()) {
//            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
//        }
//
//        //验证job是否是启用状态
//        checkJobIsEable(param.getJobId());
//
//        JobInfo jobInfo = jobInfoService.getById(param.getJobId());
//        checkJobIsMod(jobInfo.getId());
//
//        JobNodeConfig jobNodeConfig = param.getJobNodeConfig();
//        if (null!=jobNodeConfig.getExpression()&&!"null".equals(jobNodeConfig.getExpression())){
//            JSONObject minuteJson = JSONObject.parseObject(jobNodeConfig.getExpression());
//            if (null!=minuteJson.get("minute")&&!"".equals(minuteJson.get("minute"))){
//                String minute = minuteJson.get("minute").toString();
//                if(RegexUtils.checkInt(minute)){
//                    if (Integer.parseInt(minute)>59||Integer.parseInt(minute)<0){
//                        throw new RRException("请输入0-59之间的数字");
//                    }
//                } else{
//                    throw  new RRException("请输入0-59之间的数字");
//                }
//            }
//        }
//
//        //验证对应的节点配置信息是否正确,sap数据源接入不需要验证input_input_content
//        jobNodeInfoService.checkJobConfigInfo(jobNodeConfig, param.getTypeId());
//
//        //验证spark 的 resource配置，未配置则是默认值
//        if (param.getTypeId().equals(JobType.INPUT.getCode())||param.getTypeId().equals(JobType.OUTPUT.getCode())){
//            jobNodeConfig = jobNodeInfoService.checkResourceConfig(jobNodeConfig);
//        }
//
//        SapConnectField sapConnectField = jobNodeConfig.getSapConnectFieldInfo();
//        net.sf.json.JSONObject jsonObject = new net.sf.json.JSONObject();
//        //SAP的判断逻辑
//        if( null != jobNodeConfig.getInput_connect_type() && ConnectTypeEnum.Sap.getCode().equals(jobNodeConfig.getInput_connect_type())){
//            if(sapConnectField == null){
//                return R.error("sap的数据源输入参数不能为空");
//            }
//            List<String> sap_output_table_names = jobNodeConfig.getSap_output_table_name();
//            String dbname = null;
//            List<String> tableNames = new ArrayList<>();
//            if(null != sap_output_table_names){
//                if(!CollectionUtil.isEmpty(sap_output_table_names)){
//                    dbname = sap_output_table_names.get(0).substring(0,sap_output_table_names.get(0).indexOf("."));
//                }
//                for (int i = 0; i < sap_output_table_names.size() ; i++) {
//                    tableNames.add(sap_output_table_names.get(i).substring(sap_output_table_names.get(i).indexOf(".")+1));
//                }
//            }
//            //校验分区
//            if(null != dbname && tableNames.size() != 0){
//                checkSap(tableNames,dbname);
//            }
//            //获取sap配置信息,并判断是否合规
//            jobNodeConfig.setSapConnectFieldInfo(new SapCheckUtils().checkSapParams(sapConnectField));
//            jsonObject = net.sf.json.JSONObject.fromObject(sapConnectField);
//        }
//        JobNodeConfigDto jobNodeConfigDto = new JobNodeConfigDto();
//        BeanUtil.copyProperties(jobNodeConfig,jobNodeConfigDto);
//        if(jsonObject != null){
//            jobNodeConfigDto.setSapConnectFieldInfos(jsonObject.toString());
//        }
//
//        //保存或者更新job节点信息
//        jobNodeInfoService.saveOrUpdateJobNode(param);
//        JobNodeInfo nodeInfo = jobNodeInfoService.getOne(new QueryWrapper<JobNodeInfo>()
//                .eq("node_key", param.getNodeKey())
//                .eq("job_id", param.getJobId()));
//        if(!Objects.equals(param.getTypeId(),5)){
//            jobNodeConfigDto.setJobNodeId(nodeInfo.getId());
//            jobNodeConfig.setJobNodeId(nodeInfo.getId());
//            jobNodeConfig.setJobType(0);
//            jobNodeConfigDto.setJobType(0);
//            if(param.getTypeId() == 1){
//                if(null != param.getJobNodeConfig().getInput_connect_id()
//                        && !ConnectTypeEnum.Sap.getCode().equals(param.getJobNodeConfig().getInput_connect_type())
//                && !ConnectTypeEnum.WebService.getCode().equals(param.getJobNodeConfig().getInput_connect_type())){
//                    jobNodeConfService.upsertKVByJobNodeId(JobNodeConfig.jobNodeConfigTolist(jobNodeConfig));
//                }else if(null != param.getJobNodeConfig().getInput_connect_id()
//                        && (ConnectTypeEnum.Sap.getCode().equals(param.getJobNodeConfig().getInput_connect_type())
//                ||ConnectTypeEnum.WebService.getCode().equals(param.getJobNodeConfig().getInput_connect_type()))){
//                    jobNodeConfService.upsertKVByJobNodeId(JobNodeConfig.jobNodeConfigDtoTolist(jobNodeConfigDto));
//                }
//            }else{
//                jobNodeConfService.upsertKVByJobNodeId(JobNodeConfig.jobNodeConfigTolist(jobNodeConfig));
//            }
//        }
//
//        // 将shell保存至hdfs
//        Long userId = ShiroUtils.getUserId();
//        Integer jobId = param.getJobId();
//        String nodeKey = param.getNodeKey();
//        String nodeName = JobType.ObjOf(param.getTypeId()).getEnName();
//        String dst =  "workflow-app/" + userId.toString() + "/" + jobId.toString() + "/" + nodeName +  nodeKey.replace("-","") + ".sh";
//        jobNodeInfoService.uploadShell(param,dst, oozieConfig);
//
//        return R.ok();
//
//
//    }


    /**
     * 检测选中的SAP输出数仓表的分区
     */
    //@SysLog(value="检测选中的SAP输出数仓表的分区",type= OperationTypeEnum.SELECTCODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    @ApiOperation("检测选中的SAP输出数仓表的分区")
    @PostMapping("/checkSapPartition")
    public R checkSapPartition(@ApiParam(value = "sapTableNames:{'','','',''}") @RequestBody  Map<String,Object> param){

        if(null == param || param.get("sapTableNames") == null){
            return R.error("请选择SAP输出仓库表");
        }
        Set<String> partitionNames = new HashSet<>();
        checkSap(param,partitionNames);
        return R.okWithData(partitionNames);
    }


    private void checkSap(Map<String,Object> param,Set<String> partitionNames){
        List<String> tableNames = new ArrayList<>();
        String[] tableName = param.get("sapTableNames").toString().substring(1, param.get("sapTableNames").toString().length() - 1).split(",");
        List<String> list = Arrays.asList(tableName);
//        Collections.singletonList(param.get("sapTableNames").toString());
        String dbName = null;
        dbName = list.get(0).substring(0,list.get(0).indexOf("."));
        for (int i = 0; i < list.size() ; i++) {
            tableNames.add(list.get(i).substring(list.get(i).indexOf(".")+1));
        }
//        tableNames.addAll(Collections.singletonList(param.get("sapTableNames").toString()));

        List<Integer> tableInfoIds = tableInfoService.getIdsByNames(tableNames,dbName);
        //如果是kudu引擎、kudu表，则不需要配置分区
        if(CollectionUtil.isNotEmpty(tableInfoIds)){
            TableInfo tableInfo = tableInfoService.getById(tableInfoIds.get(0));
            if(null != tableInfo && Objects.equals(tableInfo.getStoreEngine(),StorageEngineEnum.KUDU.getCode())){
                //不需要填写分区
                return;
            }
        }
        List<TableFieldInfo> tableFieldInfos = new ArrayList<>();
        String tableNameWithNoPartition = "";
        boolean hasPartition = true;
        int partitonNum = 0;
        for (Integer t : tableInfoIds
        ) {
            Set<String> presentPartitionNames = new HashSet<>();
            tableFieldInfos = tableFieldInfoService.list(new QueryWrapper<TableFieldInfo>().eq("is_partition", 1).eq("table_id", t));

            if(null != tableFieldInfos && tableFieldInfos.size() != 0){
                if(!hasPartition){
                    throw new RRException(String.format("选中的输出表%s未设置分区",tableNameWithNoPartition));
                }
                if(partitonNum == 0){
                    partitonNum = tableFieldInfos.size();
                }else if(partitonNum != tableFieldInfos.size()){
                    throw new RRException("选中的输出表中分区个数不一致");
                }
                tableFieldInfos.forEach(tableFieldInfo -> presentPartitionNames.add(tableFieldInfo.getFieldName()));
                if(partitionNames.size() == 0){
                    tableFieldInfos.forEach(tableFieldInfo -> partitionNames.add(tableFieldInfo.getFieldName()));
                }else if(!checkSetSame(partitionNames,presentPartitionNames)){
                    throw new RRException("选中的SAP输出表的分区名称不同");
                }
            }else{
                TableInfo tableInfo = tableInfoService.getOne(new QueryWrapper<TableInfo>().eq("id",t));
                if(null == tableInfo){
                    throw new RRException("后台错误,未查询到数仓表格数据");
                }
                if(partitionNames.size() != 0){
                    throw new RRException(String.format("选中的输出表%s未设置分区",tableInfo.getTableName()));
                }
                hasPartition = false;
                tableNameWithNoPartition = tableInfo.getTableName();
            }
        }
    }

    private void checkSap(List<String> tableNames,String dbName){
        List<Integer> tableInfoIds = tableInfoService.getIdsByNames(tableNames,dbName);
        List<TableFieldInfo> tableFieldInfos = new ArrayList<>();
        Set<String> partitionNames = new HashSet<>();String tableNameWithNoPartition = "";
        boolean hasPartition = true;
        int partitonNum = 0;
        for (Integer t : tableInfoIds
        ) {
            Set<String> presentPartitionNames = new HashSet<>();
            tableFieldInfos = tableFieldInfoService.list(new QueryWrapper<TableFieldInfo>().eq("is_partition", 1).eq("table_id", t));

            if(null != tableFieldInfos && tableFieldInfos.size() != 0){
                if(!hasPartition){
                    throw new RRException(String.format("选中的输出表%s未设置分区",tableNameWithNoPartition));
                }
                if(partitonNum == 0){
                    partitonNum = tableFieldInfos.size();
                }else if(partitonNum != tableFieldInfos.size()){
                    throw new RRException("选中的输出表中分区个数不一致");
                }
                tableFieldInfos.forEach(tableFieldInfo -> presentPartitionNames.add(tableFieldInfo.getFieldName()));
                if(partitionNames.size() == 0){
                    tableFieldInfos.forEach(tableFieldInfo -> partitionNames.add(tableFieldInfo.getFieldName()));
                }else if(!checkSetSame(partitionNames,presentPartitionNames)){
                    throw new RRException("选中的SAP输出表的分区名称不同");
                }
            }else{
                TableInfo tableInfo = tableInfoService.getOne(new QueryWrapper<TableInfo>().eq("id",t));
                if(null == tableInfo){
                    throw new RRException("后台错误,未查询到数仓表格数据");
                }
                if(partitionNames.size() != 0){
                    throw new RRException(String.format("选中的输出表%s未设置分区",tableInfo.getTableName()));
                }
                hasPartition = false;
                tableNameWithNoPartition = tableInfo.getTableName();
            }
        }
    }

    /**
     * 检测两个set是否内容一致
     * @param set1
     * @param set2
     * @return
     */
    public Boolean checkSetSame(Set<String> set1, Set<String> set2){
        Set<String> set3 = new HashSet<>();
        set3.addAll(set1);
        if(set1.size() != set2.size()){
            return false;
        }else{
            set1.addAll(set2);
            if(set1.size() != set3.size()){
                return false;
            }
        }
        return true;
    }



//    /**
//     * 保存结束数据任务节点配置
//     */
//    @SysLog(value = "保存结束数据任务节点配置", type = OperationTypeEnum.ADDCODE, modulesType = OperationModulesEnum.JOBMODELCODE)
//    @ApiOperation("保存结束数据任务节点配置")
//    @PostMapping("/saveEndNode")
//    //    @RequiresPermissions("bdp:jobNodeInfo:save")
//    public R saveEndNode(@ApiParam(value = "{'name':'test','nodeKey':'2','jobId':1,'typeId':1,'param':''}") @RequestBody JobNodeAndParamVO jobNodeAndParamVO) throws Exception {
//
//        //验证参数的数据
//        if (null == jobNodeAndParamVO || StringUtils.isEmpty(jobNodeAndParamVO.getNodeKey()) || null == jobNodeAndParamVO.getJobId()) {
//            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
//        }
//        if(5 != jobNodeAndParamVO.getTypeId()){
//            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
//        }
//        if(StringUtils.isEmpty(jobNodeAndParamVO.getParam())){
//            throw new RRException(DataDevelopmentBizExceptionEnum.PARAM_ERROR.getMessage());
//        }
//        //验证job是否是启用状态
//        checkJobIsEable(jobNodeAndParamVO.getJobId());
//
//        //验证对应的节点配置信息是否正确
//        jobNodeInfoService.checkJobConfigInfo(null, jobNodeAndParamVO.getTypeId());
//
//        JobInfo jobInfo = jobInfoService.getById(jobNodeAndParamVO.getJobId());
//        checkJobIsMod(jobInfo.getId());
//
//        //保存或者更新job节点信息
//        JobNodeInfo param = new JobNodeInfo();
//        param.setJobId(jobNodeAndParamVO.getJobId());
//        param.setNodeKey(jobNodeAndParamVO.getNodeKey());
//        param.setTypeId(jobNodeAndParamVO.getTypeId());
//        param.setName(jobInfo.getName());
//        jobNodeInfoService.saveOrUpdateJobNode(param);
//
//        //保存和更新job信息
//        jobInfo.setParam(jobNodeAndParamVO.getParam());
//        jobInfo.setModPer(ShiroUtils.getUserId().intValue());
//        jobInfo.setModTime(DateUtil.date());
//        jobInfoService.saveOrUpdate(jobInfo);
//
//        //保存shell到hdfs
//        Long userId = ShiroUtils.getUserId();
//        Integer jobId = param.getJobId();
//        String nodeKey = param.getNodeKey();
//        String nodeName = JobType.ObjOf(param.getTypeId()).getEnName();
//        String  dst = "workflow-app/" + userId.toString() + "/" + jobId.toString() + "/" + nodeName + nodeKey.replace("-","") + ".sh";
//        jobNodeInfoService.uploadShell(param, dst, oozieConfig);
//
//        return R.ok();
//
//    }


    /**
     * 删除任务节点
     */
    @ApiOperation("删除任务节点")
    @SysLog(value="删除任务节点",type= OperationTypeEnum.DELETECODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    @PostMapping("/delete")
    //    @RequiresPermissions("bdp:jobNodeInfo:delete")
    public R delete(@ApiParam(value = "{'nodeKey':1,'jobId':1}") @RequestBody JobNodeInfo param) {
        //验证job是否为禁用，为禁用才能删除和编辑任务节点
        checkJobIsEable(param.getJobId());
        checkJobIsMod(param.getJobId());
        Map<String, Object> columnMap = new HashMap<>();
        columnMap.put("node_key", param.getNodeKey());
        columnMap.put("job_id", param.getJobId());
        JobInfo jobInfo = jobInfoService.getById(param.getJobId());
        if(null == jobInfo){
            throw new RRException("任务不存在");
        }
        //先查询对应的数据,如果，没有，则直接删除
        JobNodeInfo jobNodeInfo = jobNodeInfoService.getOne(new QueryWrapper<JobNodeInfo>().eq("job_id", param.getJobId()).eq("node_key", param.getNodeKey()));
        if(null == jobNodeInfo){
            return R.ok();
        }
        //如果是开始节点，则不能修改
        if(jobNodeInfo.getTypeId() == 0){
            throw new RRException("开始节点不能删除！");
        }

        if (jobNodeInfoService.removeByMap(columnMap)) {
                //String.valueOf(jobInfo.getId()) :坑根据工程路径进行删除
                String folderJobPath = jobInfo.getProcId() +"/"+ jobInfo.getId();
                jobNodeConfService.remove(new QueryWrapper<JobNodeConf>().eq("job_node_id",jobNodeInfo.getId()));
                jobNodeRunHistoryService.remove(new QueryWrapper<JobNodeRunHistory>().eq("job_node_info_id",jobNodeInfo.getId()));

            //删除hdfs上的数据
            try{
                jobNodeInfoService.deleteHdfsInfo(jobNodeInfo);
                jobNodeInfoService.deleteHdfsShellFile(jobInfo,jobNodeInfo);
            }catch (Exception e){

            }
            return R.ok();
        }
        return R.ok();
    }


    private void checkJobIsEable(Integer jobId) {
        if (!jobInfoService.checkJobIsEable(jobId)) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOB_INFO_NO_EABLE.getMessage());
        }
    }

    private void checkJobIsMod(Integer jobId) {
        if (!jobInfoService.checkExistById(jobId)) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOB_INFO_NOTEXISTED.getMessage());
        }
        if (!jobInfoService.checkPermission(jobId)) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOBINFO_PERMISSIOIN.getMessage());
        }
        if (jobInfoService.checkIsRun(jobId)) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOBINFO_RUN.getMessage());
        }
    }

    /**
     * 上传文件
     */
    @ApiOperation("上传文件")
    @SysLog(value="上传文件",type= OperationTypeEnum.SELECTCODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    @PostMapping("/upload")
    //    @RequiresPermissions("bdp:jobNodeInfo:upload")
    public R upload(@RequestParam("file") MultipartFile file, Integer jobId) {
        JobInfo jobInfo = jobInfoService.getById(jobId);
        if (jobInfo.getCreatePer() != ShiroUtils.getUserId().intValue()) {
            throw new RRException(DataDevelopmentBizExceptionEnum.JOBINFO_PERMISSIOIN.getMessage());
        }
        if (!file.isEmpty()) {
            HdfsUtil util = null;
            try {
                util = new HdfsUtil(bdpJobConfig);
                util.writeFile(file.getBytes(), String.format("%s/%s/%s", bdpJobConfig.getJoblib(), jobId, file.getOriginalFilename()));
            } catch (Exception e) {
                throw new RRException(e.getMessage(), 500);
            }finally {
                if(null != util){
                    util.close();
                }
            }
            return R.ok().put("file", file.getOriginalFilename());
        } else {
            throw new RRException("文件为空!!!", 500);
        }
    }

    /**
     * 获取所有的循环类别
     */
    @ApiOperation("获取所有的循环类别")
    //@SysLog(value="获取所有的循环类别",type= OperationTypeEnum.SELECTCODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    @GetMapping("/listSchedules")
//    @RequiresPermissions("bdp:jobNodeInfo:listSchedules")
    public R listSchedules() {
        Map<String,List> data = new HashMap<>();
        List<JSONObject> runStyle = new ArrayList<>();
        List<JSONObject> scheduleStyle = new ArrayList<>();
        for (ScheduleType e : ScheduleType.values()
        ) {
            JSONObject obj = new JSONObject();
            obj.put("id", e.getCode());
            obj.put("name", e.getName());
            runStyle.add(obj);
        }
        for (ScheduleRepeatType e : ScheduleRepeatType.values()
        ) {
            JSONObject obj = new JSONObject();
            obj.put("id", e.getCode());
            obj.put("name", e.getName());
            scheduleStyle.add(obj);
        }
        data.put("runStyle",runStyle);
        data.put("scheduleStyle",scheduleStyle);
        return R.okWithData(data);
    }

    /**
     * 任务节点详情
     */
    @ApiOperation("任务节点详情")
    @SysLog(value="任务节点详情",type= OperationTypeEnum.SELECTCODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    @GetMapping("/info/{id}")
//    @RequiresPermissions("bdp:jobNodeInfo:info")
    public R info(@PathVariable("id") Integer id) {
        JobNodeInfo jobNodeInfo = jobNodeInfoService.getById(id);
        JobNodeConfig jobNodeConfig = new JobNodeConfig();
        R r = R.ok();
        if (jobNodeInfo != null) {
            List<JobNodeConf> jobNodeConfs = jobNodeConfService.list(new QueryWrapper<JobNodeConf>().eq("job_node_id", jobNodeInfo.getId()).eq("job_type",0));
            jobNodeConfig = JobNodeConfig.listToJobNodeConfig(jobNodeConfs);
            //添加默认值
            addFtpDefaults(jobNodeConfig,jobNodeConfs);
            r.put("data",jobNodeConfig);
            //对应的任务节点对应的
            List<JobNodeLayConfigDto> reList = jobNodeInfoService.selectJobNodeLayList(jobNodeInfo);
            r.put("jobNodeLayConfigList",reList);
            //查询到责任人信息
            SysUserEntity sysUserEntity = sysUserService.getById(jobNodeInfo.getCreatePer());
            if(null != sysUserEntity){
                jobNodeInfo.setCreatePerName(sysUserEntity.getUsername());
            }
            r.put("jobNodeInfo",jobNodeInfo);
        }
        return r;
    }

    private void addFtpDefaults(JobNodeConfig jobNodeConfig, List<JobNodeConf> jobNodeConfs) {
        if(jobNodeConfig != null){
            if(CollectionUtil.isNotEmpty(jobNodeConfs)){
                Set<String> keys = jobNodeConfs.stream().map(JobNodeConf::getKey).collect(Collectors.toSet());
                if(!keys.contains("ftpFileMatchType")){
                    jobNodeConfig.setFtpFileMatchType(1);
                }
                if(!keys.contains("ftpFileType")){
                    jobNodeConfig.setFtpFileType("json");
                }
            }
        }
    }

    private void addFtpDefaults(JobNodeConfigDto jobNodeConfig, List<JobNodeConf> jobNodeConfs) {
        if(jobNodeConfig != null){
            if(CollectionUtil.isNotEmpty(jobNodeConfs)){
                Set<String> keys = jobNodeConfs.stream().map(JobNodeConf::getKey).collect(Collectors.toSet());
                if(!keys.contains("ftpFileMatchType")){
                    jobNodeConfig.setFtpFileMatchType(1);
                }
                if(!keys.contains("ftpFileType")){
                    jobNodeConfig.setFtpFileType("json");
                }
            }
        }
    }

    /**
     * 任务节点详情
     */
    @ApiOperation("任务节点配置详情")
    //@SysLog(value="任务节点配置详情",type= OperationTypeEnum.SELECTCODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    @GetMapping("/infoByKeyAndJobId/{jobNodeId}")
//    @RequiresPermissions("bdp:jobNodeInfo:infoByKeyAndJobId")
    public R infoByKeyAndJobId(@PathVariable("jobNodeId") Integer jobNodeId) {
        JobNodeInfo jobNodeInfo = jobNodeInfoService.getById(jobNodeId);
        JobNodeConfigVo jobNodeConfigVo = new JobNodeConfigVo();
        if (jobNodeInfo != null) {
            //查询到责任人信息
            SysUserEntity sysUserEntity = sysUserService.getById(jobNodeInfo.getCreatePer());
            if(null != sysUserEntity){
                jobNodeInfo.setCreatePerName(sysUserEntity.getUsername());
            }
            if(Objects.equals(jobNodeInfo.getTypeId(),JobType.SHELL.getCode())){
                //shell的端口和host
                BussessConfigEntity bussessConfigEntity = bussessConfigService.getOne(new QueryWrapper<BussessConfigEntity>()
                        .eq("bussess_type", 1).eq("bussess_id", jobNodeId));
                if(null != bussessConfigEntity){
                    jobNodeInfo.setPort(bussessConfigEntity.getPort());
                    jobNodeInfo.setHostUrl(bussessConfigEntity.getHostUrl());
                    jobNodeInfo.setUsername(bussessConfigEntity.getUsername());
                }
            }
            jobNodeConfigVo.setJobNodeInfo(jobNodeInfo);
            //查询到任务的工程id
            JobInfo jobInfo = jobInfoService.getById(jobNodeInfo.getJobId());
            if(null != jobInfo){
                jobNodeConfigVo.setProcId(jobInfo.getProcId());
            }

            List<JobNodeConf> jobNodeConfs = jobNodeConfService.list(new QueryWrapper<JobNodeConf>()
                    .eq("job_node_id", jobNodeInfo.getId())
            .eq("job_type",0));
            JobNodeConfigDto jobNodeConfigDto = jobNodeConfService.getJobNodeConf(jobNodeConfs);
            //添加默认值
            addFtpDefaults(jobNodeConfigDto,jobNodeConfs);

            //处理表输出 jobOutPutTableIds
            if(null != jobNodeConfigDto && null != jobNodeConfigDto.getJobOutPutTableIds() && jobNodeConfigDto.getJobOutPutTableIds().size() > 0){
                List<String> jobOutPutTableIds = jobNodeConfigDto.getJobOutPutTableIds();
                for (String jobOutPutTableId : jobOutPutTableIds) {
                    String sapNameSub = jobOutPutTableId.substring(1, jobOutPutTableId.length() - 1);
                    String[] sapArray = sapNameSub.split(",");
                    List<String> sapList = Arrays.asList(sapArray);
                    List<List<Object>> outputTableNameList = new ArrayList<>();
                    for(int i = 0;i<sapList.size();i++){
                        String s = sapList.get(i);
                        if(StringUtils.isNotEmpty(s)){
                            s = s.replace("\"","");
                            TableInfo tableInfo = getTableInfo(s);
                            if(null != tableInfo){
                                List<Object> objects = new ArrayList<>();
                                objects.add(tableInfo.getDbName());
                                objects.add(Integer.valueOf(s.trim()));
                                outputTableNameList.add(objects);
                            }
                        }
                    }
                    jobNodeConfigVo.setJobOut(outputTableNameList);
                }
            }

            List<Map<String,Object>> sap_output_table_name = new ArrayList<>();
            List<String> sap_output_table_name_older = jobNodeConfigDto.getSap_output_table_name();
            //获取此jobid配置的数据源是否是SAP数据源，若是则解析数据；
            Integer input_connect_type = jobNodeConfigDto.getInput_connect_type();
            JSONObject jsonObject = new JSONObject();
            if(ConnectTypeEnum.Sap.getCode().equals(input_connect_type)
                    ||ConnectTypeEnum.WebService.getCode().equals(input_connect_type)){
                //获取SAP输出表数据
                if(null != sap_output_table_name_older && sap_output_table_name_older.size() != 0){
                    for (String sapName : sap_output_table_name_older) {
                        String sapNameSub = sapName.substring(1, sapName.length() - 1);
                        String[] sapArray = sapNameSub.split(",");
                        List<String> sapList = Arrays.asList(sapArray);
                        for (String s : sapList) {
                            if (!s.trim().equals("")) {
                                Map<String, Object> sapOutputTableNameMap = new HashMap<>();
                                sapOutputTableNameMap.put("tableName", s.trim().substring(s.trim().indexOf(".")+1));
                                sap_output_table_name.add(sapOutputTableNameMap);
                            }
                        }
                    }
                }

                //获取SAP数据源配置（和最开始的数据源不同，是单独保存的配置信息）
                String sapConnectFieldInfos = jobNodeConfigDto.getSapConnectFieldInfos();
                if(sapConnectFieldInfos != null) {
                    List<String> sapParams = new ArrayList<>();
                    if (ConnectTypeEnum.Sap.getCode().equals(input_connect_type)) {
                        sapParams.addAll(Arrays.asList("fieldName", "fieldKind", "fieldValue", "fieldAlias", "fieldInputType", "fieldDataType"));
                    }
                    else if (ConnectTypeEnum.WebService.getCode().equals(input_connect_type)){
                        sapParams.addAll(Arrays.asList("fieldName", "fieldValue", "fieldAlias"));
                    }
                    jsonObject = JSONObject.parseObject(sapConnectFieldInfos);
                    if (jsonObject != null && jsonObject.containsKey("ainputParams")) {
                        String ainputParams = jsonObject.get("ainputParams").toString();
                        JSONArray jsonArray = JSONArray.parseArray(ainputParams);
                        JSONArray jsonArray1 = new JSONArray();
                        for (Object obj : jsonArray
                        ) {
                            String string = JSONObject.toJSONString(obj);
                            JSONObject jsonObject1 = JSONObject.parseObject(string);
                            List<String> keys = new ArrayList<>(jsonObject1.keySet());
                            sapParams.removeAll(keys);
                            for (String s : sapParams
                            ) {
                                jsonObject1.put(s, "");
                            }
                            jsonArray1.add(jsonObject1);
                        }
                        jsonObject.put("ainputParams", jsonArray1);
                    }
                }
            }else{
                //其它的转换实体会报错
                jobNodeConfigDto.setSap_output_table_name(null);
            }

            //解析数据分区成前端数据格式
            String output_data_partition_older = jobNodeConfigDto.getOutput_data_partition();
            List<Map<String,String>> output_data_partition = new ArrayList<>();
            if(output_data_partition_older != null && ! "".equals(output_data_partition_older)){
                String[] output_data_partitions = output_data_partition_older.split(",");
                for (String s : output_data_partitions
                     ) {
                    if(!s.contains("=")){
                        continue;
                    }
                    Map<String,String> presentMap = new HashMap<>();
                    String fieldName = s.substring(0,s.indexOf("="));
                    String fieldValue = s.substring(s.indexOf("=")+1);
                    presentMap.put("fieldName",fieldName);
                    presentMap.put("fieldValue",fieldValue);
                    output_data_partition.add(presentMap);
                }
            }
            BeanUtils.copyProperties(jobNodeConfigDto,jobNodeConfigVo);
            jobNodeConfigVo.setSap_output_table_name(sap_output_table_name);
            jobNodeConfigVo.setOutput_data_partitions(output_data_partition);
            //解析算法开发任务preResultInfos字段
            if(JobType.ALGORITHM.getCode() == jobNodeInfo.getTypeId()){
                String preResultInfos = jobNodeConfigDto.getPreResultInfos();
//                Object parse = JSONObject.parseArray(preResultInfos, List.class);
                List<Map<String, Object>> preResultInfosList = JSON.parseObject(preResultInfos, new TypeReference<List<Map<String, Object>>>() {
                });
                jobNodeConfigVo.setPreResultInfos(preResultInfosList);
            }
            if(jsonObject.size() != 0){
                SapConnectField sapConnectField = mapToSapConnectField(jsonObject);
                jobNodeConfigVo.setSapConnectFieldInfo(sapConnectField);
            }
            //节点依赖路由
            //对应的任务节点对应的
            List<JobNodeLayConfigDto> reList = jobNodeInfoService.selectJobNodeLayList(jobNodeInfo);
            jobNodeConfigVo.setJobNodeLayConfigDtos(reList);
        }
        return R.okWithData(jobNodeConfigVo);
    }



    /**
     * 通过关联表名称获取任务信息
     */
    @ApiOperation("通过关联表名称获取任务信息")
    //@SysLog(value="通过关联表名称获取任务信息",type= OperationTypeEnum.SELECTCODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    @PostMapping("/getJobDataPage")
    public R getJobDataPage(@RequestBody JobDataVo jobDataVo) {
        Page<JobDataRes> dataResList = jobNodeConfService.getJobDataPage(jobDataVo);
        return R.okWithData(dataResList);
    }



    /**
     * 通过关联表名称获取任务创建人信息
     */
    @ApiOperation("通过关联表名称获取任务创建人信息")
    @PostMapping("/getJobDataCreateBy")
    public R getJobDataCreateBy(@RequestBody JobDataVo jobDataVo) {
        List<Map<String,Object>> dataResList = jobNodeConfService.getJobDataCreateBy(jobDataVo);
        return R.okWithData(dataResList);
    }



    /**
     * 补录关联的任务
     */
    @ApiOperation("补录关联的任务")
    //@SysLog(value="补录关联的任务",type= OperationTypeEnum.SELECTCODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
    @PostMapping("/insertJobNodeTable")
    public R insertJobNodeTable(@RequestBody JobNodeTable jobNodeTable) {
        jobNodeConfService.insertJobNodeTable(jobNodeTable);
        return R.ok();
    }


//    /**
//     * 任务节点详情
//     */
//    @ApiOperation("复制任务节点配置详情")
//    //@SysLog(value="任务节点配置详情",type= OperationTypeEnum.SELECTCODE, modulesType = OperationModulesEnum.BDPMODELCODE,platformModuleType = ModuleEnum.PROJECT_LIST_MODULE_CODE)
//    @GetMapping("/copyInfoByKeyAndJobId/{jobNodeId}")
//    public R copyInfoByKeyAndJobId(@PathVariable("jobNodeId") Integer jobNodeId) {
//        JobNodeInfo jobNodeInfo = jobNodeInfoService.getById(jobNodeId);
//        JobNodeConfigVo jobNodeConfigVo = new JobNodeConfigVo();
//        if (jobNodeInfo != null) {
//            //查询到责任人信息(暂时不需要)
//            if(Objects.equals(jobNodeInfo.getTypeId(),JobType.SHELL.getCode())){
//                //shell的端口和host
//                BussessConfigEntity bussessConfigEntity = bussessConfigService.getOne(new QueryWrapper<BussessConfigEntity>()
//                        .eq("bussess_type", 1).eq("bussess_id", jobNodeId));
//                if(null != bussessConfigEntity){
//                    jobNodeInfo.setPort(bussessConfigEntity.getPort());
//                    jobNodeInfo.setHostUrl(bussessConfigEntity.getHostUrl());
//                    jobNodeInfo.setUsername(bussessConfigEntity.getUsername());
//                }
//            }
//            jobNodeConfigVo.setJobNodeInfo(jobNodeInfo);
//            //查询到任务的工程id
//            JobInfo jobInfo = jobInfoService.getById(jobNodeInfo.getJobId());
//            if(null != jobInfo){
//                jobNodeConfigVo.setProcId(jobInfo.getProcId());
//            }
//            //将需要复制的数据全部查询出来。并且转换成为对象返回
//            List<JobNodeConf> jobNodeConfs = jobNodeConfService.list(new QueryWrapper<JobNodeConf>()
//                    .eq("job_node_id", jobNodeId)
//                    .eq("job_type",0)
//                    .ne("`key`","priorityConfig")
//                    .ne("`key`","jobOutPutTableIds")
//                    .ne("`key`","jobNodeId"));
//            JobNodeConfigDto jobNodeConfigDto = jobNodeConfService.getJobNodeConf(jobNodeConfs);
//            List<Map<String,Object>> sap_output_table_name = new ArrayList<>();
//            List<String> sap_output_table_name_older = jobNodeConfigDto.getSap_output_table_name();
//            //获取此jobid配置的数据源是否是SAP数据源，若是则解析数据；
//            Integer input_connect_type = jobNodeConfigDto.getInput_connect_type();
//            JSONObject jsonObject = new JSONObject();
//            if(ConnectTypeEnum.Sap.getCode().equals(input_connect_type)
//                    ||ConnectTypeEnum.WebService.getCode().equals(input_connect_type)){
//                //获取SAP输出表数据
//                if(null != sap_output_table_name_older && sap_output_table_name_older.size() != 0){
//                    for (String sapName : sap_output_table_name_older) {
//                        String sapNameSub = sapName.substring(1, sapName.length() - 1);
//                        String[] sapArray = sapNameSub.split(",");
//                        List<String> sapList = Arrays.asList(sapArray);
//                        for (String s : sapList) {
//                            if (!s.trim().equals("")) {
//                                Map<String, Object> sapOutputTableNameMap = new HashMap<>();
//                                sapOutputTableNameMap.put("tableName", s.trim().substring(s.trim().indexOf(".")+1));
//                                sap_output_table_name.add(sapOutputTableNameMap);
//                            }
//                        }
//                    }
//                }
//
//                //获取SAP数据源配置（和最开始的数据源不同，是单独保存的配置信息）
//                String sapConnectFieldInfos = jobNodeConfigDto.getSapConnectFieldInfos();
//                if(sapConnectFieldInfos != null) {
//                    List<String> sapParams = new ArrayList<>();
//                    if (ConnectTypeEnum.Sap.getCode().equals(input_connect_type)) {
//                        sapParams.addAll(Arrays.asList("fieldName", "fieldKind", "fieldValue", "fieldAlias", "fieldInputType", "fieldDataType"));
//                    }
//                    else if (ConnectTypeEnum.WebService.getCode().equals(input_connect_type)){
//                        sapParams.addAll(Arrays.asList("fieldName", "fieldValue", "fieldAlias"));
//                    }
//                    jsonObject = JSONObject.parseObject(sapConnectFieldInfos);
//                    if (jsonObject != null && jsonObject.containsKey("ainputParams")) {
//                        String ainputParams = jsonObject.get("ainputParams").toString();
//                        JSONArray jsonArray = JSONArray.parseArray(ainputParams);
//                        JSONArray jsonArray1 = new JSONArray();
//                        for (Object obj : jsonArray
//                        ) {
//                            String string = JSONObject.toJSONString(obj);
//                            JSONObject jsonObject1 = JSONObject.parseObject(string);
//                            List<String> keys = new ArrayList<>(jsonObject1.keySet());
//                            sapParams.removeAll(keys);
//                            for (String s : sapParams
//                            ) {
//                                jsonObject1.put(s, "");
//                            }
//                            jsonArray1.add(jsonObject1);
//                        }
//                        jsonObject.put("ainputParams", jsonArray1);
//                    }
//                }
//            }else{
//                //其它的转换实体会报错
//                jobNodeConfigDto.setSap_output_table_name(null);
//            }
//
//            //解析数据分区成前端数据格式
//            String output_data_partition_older = jobNodeConfigDto.getOutput_data_partition();
//            List<Map<String,String>> output_data_partition = new ArrayList<>();
//            if(output_data_partition_older != null && ! "".equals(output_data_partition_older)){
//                String[] output_data_partitions = output_data_partition_older.split(",");
//                for (String s : output_data_partitions
//                ) {
//                    if(!s.contains("=")){
//                        continue;
//                    }
//                    Map<String,String> presentMap = new HashMap<>();
//                    String fieldName = s.substring(0,s.indexOf("="));
//                    String fieldValue = s.substring(s.indexOf("=")+1);
//                    presentMap.put("fieldName",fieldName);
//                    presentMap.put("fieldValue",fieldValue);
//                    output_data_partition.add(presentMap);
//                }
//            }
//            BeanUtils.copyProperties(jobNodeConfigDto,jobNodeConfigVo);
//            jobNodeConfigVo.setSap_output_table_name(sap_output_table_name);
//            jobNodeConfigVo.setOutput_data_partitions(output_data_partition);
//            //解析算法开发任务preResultInfos字段
//            if(JobType.ALGORITHM.getCode() == jobNodeInfo.getTypeId()){
//                String preResultInfos = jobNodeConfigDto.getPreResultInfos();
////                Object parse = JSONObject.parseArray(preResultInfos, List.class);
//                List<Map<String, Object>> preResultInfosList = JSON.parseObject(preResultInfos, new TypeReference<List<Map<String, Object>>>() {
//                });
//                jobNodeConfigVo.setPreResultInfos(preResultInfosList);
//            }
//            if(jsonObject.size() != 0){
//                SapConnectField sapConnectField = mapToSapConnectField(jsonObject);
//                jobNodeConfigVo.setSapConnectFieldInfo(sapConnectField);
//            }
//        }
//        return R.okWithData(jobNodeConfigVo);
//    }

    private TableInfo getTableInfo(String s) {
        if(StringUtils.isNotEmpty(s)){
            try {
                return  tableInfoService.getById(Integer.valueOf(s.trim()));
            }catch (Exception e){
                log.error("错误数据格式s:{}，error:{}",s,e.getMessage());
            }
        }
        return null;
    }

    public static SapConnectField mapToSapConnectField(Map<String, Object> map) {
        if (map == null)
            return null;
        SapConnectField obj;
        try {
            JSONObject jsonObject = JSONObject.parseObject(JSON.toJSONString(map));
            obj = JSON.parseObject(jsonObject.toJSONString(), new TypeReference<SapConnectField>(){});
            return obj;
        } catch (Exception e) {
            return null;
        }
    }
}

